package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/sirupsen/logrus"
	"gopkg.in/ini.v1"
	"logagent/kafka"
	"logagent/tailfile"
	"strings"
	"time"
)

//整个Logagent的配置结构体
type Config struct {
	KafkaConfig   `ini:"kafka"`
	CollectConfig `ini:"collect"`
}

type KafkaConfig struct {
	Address  string `ini:"address"`
	ChanSize int64  `ini:"chansize"`
}

type CollectConfig struct {
	LogfilePath string `ini:"logfile_path"`
}

//真正的业务逻辑
func run() (err error) {
	//3.1从TailObj读取数据 TailObj->log->Client->kafka
	for {
		//循环读数据
		line, ok := <-tailfile.TailObj.Lines //chan tailfile.Line
		if !ok {
			logrus.Warning("tailfile file close reopen,filename: \n", tailfile.TailObj.Filename)
			time.Sleep(time.Second) //读取出错，等一秒
			continue
		}
		//如果是空行就略过
		fmt.Printf("%#v \n", line.Text)
		if len(strings.Trim(line.Text, "\r")) == 0 {
			logrus.Info("出现空行，直接跳过")
			continue
		}
		//fmt.Println("msg:", line.Text)
		//利用通道将同步的代码改为异步的
		//把读出来的一行日志包装成kafka里面的msg类型
		msg := &sarama.ProducerMessage{}
		msg.Topic = "web_log"
		msg.Value = sarama.StringEncoder(line.Text)
		//丢到通道中
		kafka.ToMsgChan(msg)
	}
	return
}
func main() {

	//0.第一种方式:读配置文件
	//cfg, err := ini.Load("./conf/config.ini")

	//0.第二种方式:读配置文件
	var configObj = new(Config)
	err := ini.MapTo(configObj, "./conf/config.ini")
	//0.1
	if err != nil {
		logrus.Error("load config failed,err:", err)
		return
	}
	fmt.Printf("%#v \n", configObj)
	//1.初始化连接kafka（做好准备工作）
	err = kafka.Init([]string{configObj.Address}, configObj.ChanSize)
	fmt.Printf("%#v \n", configObj.Address, configObj.ChanSize)
	if err != nil {
		logrus.Error("init kafka failed,err:", err)
		return
	}
	logrus.Error("init kafka success!!!")
	//2.根据配置中的日志路径使用tail去收集日志,初始化tail
	err = tailfile.Init(configObj.LogfilePath)
	fmt.Printf("%#v \n", configObj.LogfilePath)
	if err != nil {
		logrus.Error("init tailfile failed,err:", err)
		return
	}
	logrus.Error("init tailfile success!!!")
	//3.把日志通过sarama发往kafka
	//3.1从TailObj读取数据 TailObj->log->Client->kafka
	err = run()
	if err != nil {
		logrus.Error("run failed,err:", err)
		return
	}
}
